package flowprocess

import (
	"context"
	"log"
)

type Task interface{}

type TaskChan chan Task
type InTaskChan <-chan Task
type OutTaskChan chan<- Task

type Flow interface {
	//添加流处理节点。每个Processor并行执行，此方法开放性好。
	AddNodeProcessors(processorChanBufferSize int, processors ...Processor)
	//添加流处理节点。每个TaskHandler并行执行，当需要追踪任务执行轨迹时，可以使用该方法。
	AddNodeTaskHandlers(processorChanBufferSize int, taskHandlers ...TaskHandler)
	//启动。
	Start()
	//等待执行完成。
	Await()
	//等待执行结果。该方法会阻塞，直到有结果输出。当ok为true时，结果有效。
	//当任务的执行结果有多个时，可循环调用该方法获取结果。
	Result() (result interface{}, ok bool)
}

// node1(process) --result1--> node2(process) --result2--> node3(process) and so on.
type flow struct {
	nodes      []Node
	resultChan <-chan Task

	taskAt           func(task TraceableTask, nodeId int)
	onNewTaskCreated func(task TraceableTask, nodeId int)
	onTaskFinished   func(task TraceableTask, nodeId int, err error)
}

//taskAt： 当TraceableTask到达某节点时，回调该方法。
//onNewTaskCreated： 当新的TraceableTask产生时，回调该方法。
//onTaskFinished： 当onTaskFinished完成时，回调该方法。
//只有通过Flow.AddNodeTaskHandlers创建流节点，且task为TraceableTask时，才会回调上述方法。
func NewFlow(taskAt func(task TraceableTask, nodeId int),
	onNewTaskCreated func(task TraceableTask, nodeId int),
	onTaskFinished func(task TraceableTask, nodeId int, err error)) Flow {
	f := &flow{
		nodes:            []Node{},
		resultChan:       make(<-chan Task),
		taskAt:           taskAt,
		onNewTaskCreated: onNewTaskCreated,
		onTaskFinished:   onTaskFinished,
	}
	if f.taskAt == nil {
		f.taskAt = func(task TraceableTask, nodeId int) {
		}
	}
	if f.onNewTaskCreated == nil {
		f.onNewTaskCreated = func(task TraceableTask, nodeId int) {
		}
	}
	if f.onTaskFinished == nil {
		f.onTaskFinished = func(task TraceableTask, nodeId int, err error) {
		}
	}

	return f
}

func (f *flow) addNode(node Node) {
	f.nodes = append(f.nodes, node)
}

func (f *flow) AddNodeProcessors(processorChanBufferSize int, processors ...Processor) {
	f.addNode(NewNode(processors, processorChanBufferSize, len(f.nodes)))
}

func (f *flow) AddNodeTaskHandlers(processorChanBufferSize int, taskHandlers ...TaskHandler) {
	handler := NewTaskHandlerProcessors(f.taskAt, f.onNewTaskCreated, f.onTaskFinished, taskHandlers...)
	f.addNode(NewNode(handler, processorChanBufferSize, len(f.nodes)))
}

//start flow process
func (f *flow) Start() {
	var taskChan TaskChan
	for _, node := range f.nodes {
		taskChan = node.Process(taskChan)
	}
	f.resultChan = taskChan
}

func (f *flow) Await() {
	for _, pu := range f.nodes {
		<-pu.Done()
	}
}

// func (f *flow) Result() (resultChan <-chan Task) {
// 	return f.resultChan
// }

func (f *flow) Result() (result interface{}, ok bool) {
	result, ok = <-f.resultChan
	if ok {
		traceableTask, traceable := result.(TraceableTask)
		if traceable {
			f.onTaskFinished(traceableTask, len(f.nodes)-1, nil)
		}
	}
	return
}

//a node to process tasks
type Node interface {
	Process(in TaskChan) (out TaskChan)
	Done() <-chan struct{}
}

//process task from the previous FlowNode, and place the result into outTask chan which will be proccessed by the next FlowNode.
//A node includes multiple processors which run concurrently.
type Processor interface {
	Proccess(inTasks InTaskChan, outTask OutTaskChan, ctx context.Context) (cancelAllProcess bool)
}

type DefaultProcessor struct {
	handleAndDispatch TaskHandler
	taskAt            func(task TraceableTask, nodeId int)
	onNewTaskCreated  func(task TraceableTask, nodeId int)
	onTaskFinished    func(task TraceableTask, nodeId int, err error)
}

func (p *DefaultProcessor) Proccess(inTaskChan InTaskChan, outTaskChan OutTaskChan, ctx context.Context) (cancelAllProcess bool) {
	nodeId := ctx.Value(NODE_ID).(int)

	if inTaskChan == nil {
		err := p.handleAndDispatch.Handle(nil, func(outTask Task) {
			traceableOutTask, outTaskTraceable := outTask.(TraceableTask)
			p.tryTraceCreateTask(outTaskTraceable, traceableOutTask, nodeId)
			outTaskChan <- outTask
		})
		if err != nil {
			log.Printf("HandleAndDispatch error when inTaskChan is nil: %v. Begin canceling all processes.", err)
			return true
		}
		return false
	}

	for {
		select {
		case <-ctx.Done():
			return true
		case task, ok := <-inTaskChan:
			var traceableTask TraceableTask
			var traceable bool
			if ok {
				traceableTask, traceable = task.(TraceableTask)
				if traceable {
					p.taskAt(traceableTask, nodeId)
				}
			}
			thisTaskDispatched := false
			dispatchFunc := func(outTask Task) {
				thisTaskDispatched = true
				traceableOutTask, outTaskTraceable := outTask.(TraceableTask)

				if outTaskTraceable && traceable {
					if traceableTask.TaskId() != traceableOutTask.TaskId() {
						p.onTaskFinished(traceableTask, nodeId, nil)
						p.onNewTaskCreated(traceableOutTask, nodeId)
					}
				} else {
					p.tryTraceFinishedTask(traceable, traceableTask, nodeId, nil)
					p.tryTraceCreateTask(outTaskTraceable, traceableOutTask, nodeId)
				}

				outTaskChan <- outTask
			}
			var err error
			if ok {
				err = p.handleAndDispatch.Handle(task, dispatchFunc)
			} else {
				err = p.handleAndDispatch.OnCompleted(dispatchFunc)
			}
			if err != nil {
				p.tryTraceFinishedTask(traceable, traceableTask, nodeId, err)
				log.Printf("HandleAndDispatch error: %v. Begin canceling all processes.", err)
				return true
			}
			if !thisTaskDispatched {
				p.tryTraceFinishedTask(traceable, traceableTask, nodeId, nil)
			}
			if !ok {
				return false
			}
		}
	}
}

func (p *DefaultProcessor) tryTraceFinishedTask(traceable bool, traceableTask TraceableTask, nodeId int, err error) {
	if traceable {
		p.onTaskFinished(traceableTask, nodeId, err)
	}
}

func (p *DefaultProcessor) tryTraceCreateTask(traceable bool, traceableTask TraceableTask, nodeId int) {
	if traceable {
		p.onNewTaskCreated(traceableTask, nodeId)
	}
}

func NewTaskHandlerProcessors(
	taskAt func(task TraceableTask, nodeId int),
	onNewTaskCreated func(task TraceableTask, nodeId int),
	onTaskFinished func(task TraceableTask, nodeId int, err error),
	taskhandlers ...TaskHandler) []Processor {
	ps := make([]Processor, len(taskhandlers))

	for i := 0; i < len(taskhandlers); i++ {
		ps[i] = &DefaultProcessor{
			handleAndDispatch: taskhandlers[i],
			taskAt:            taskAt,
			onNewTaskCreated:  onNewTaskCreated,
			onTaskFinished:    onTaskFinished,
		}
	}
	return ps
}
